package com.zjc.tcploadbalance;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.zjc.tcploadbalance.config.InstanceConfig;
import com.zjc.tcploadbalance.config.ZookeeperConfig;
import com.zjc.tcploadbalance.constant.Constants;
import com.zjc.tcploadbalance.listener.CustomConnectionStateListener;
import com.zjc.tcploadbalance.listener.InstanceListener;
import com.zjc.tcploadbalance.model.InstanceModel;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.Environment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;

public class RegisterClient implements EnvironmentAware {

    private static final Logger logger = LoggerFactory.getLogger(RegisterClient.class);

    private static Gson gson = new GsonBuilder().create();
    private Environment environment;

    private CuratorFramework client;
    private String registerPath;

    private ZookeeperConfig zookeeperConfig;
    private InstanceConfig instanceConfig;

    private List<InstanceModel> instanceModels = new ArrayList<>();

    public RegisterClient(ZookeeperConfig zookeeperConfig, InstanceConfig instanceConfig) {
        this.zookeeperConfig = zookeeperConfig;
        this.instanceConfig = instanceConfig;
    }

    @PostConstruct
    public void autoRegister() {
        if (ConfigurableEnvironment.class.isInstance(environment)) {
            ConfigurableEnvironment configEnv = (ConfigurableEnvironment) environment;
            MapPropertySource mapPropertySource = (MapPropertySource) configEnv.getPropertySources().get("tcploadbalance");
            if (null != mapPropertySource) {
                Boolean autoRegister = (Boolean) mapPropertySource.getProperty("tcploadbalance.auto-registration.enabled");
                if (autoRegister && autoRegister) {
                    init();
                }
            }

        }
    }

    public void init() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(zookeeperConfig.getConnectString())
                .sessionTimeoutMs(zookeeperConfig.getSessionTimeoutMs())  // 会话超时时间
                .connectionTimeoutMs(zookeeperConfig.getConnectionTimeoutMs()) // 连接超时时间
                .retryPolicy(retryPolicy)
                .namespace(zookeeperConfig.getNamespace()) // 包含隔离名称
                .build();
        client.start();
        addConnectionStateListener();
        addInstanceListener();
        registerInstance();
    }


    /**
     * 创建临时节点
     *
     * @param path
     * @param content
     * @return
     * @throws Exception
     */
    public String createEphemeralSequentialNode(String path, String content) throws Exception {
        return client.create().creatingParentContainersIfNeeded() // 递归创建所需父节点
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL) // 创建类型为持久节点
                .forPath(path, content.getBytes()); // 目录及内容

    }

    public void deleteNode(String path) throws Exception {
        client.delete().deletingChildrenIfNeeded().forPath(path);
    }

    public void addConnectionStateListener() {
        CustomConnectionStateListener customConnectionStateListener = new CustomConnectionStateListener(this);
        this.client.getConnectionStateListenable().addListener(customConnectionStateListener);
    }

    public void reRegisterInstance() {
        try {
            if (StringUtils.hasText(registerPath)) {
                try {
                    if (this.isNodeExist(registerPath)) {
                        this.deleteNode(registerPath);
                    }
                } catch (Exception e) {
                    logger.error("reRegisterInstance isNodeExist/deleteNode Exception : ", e);
                }
                registerPath = this.createEphemeralSequentialNode(Constants.INSTANCE_INSTANCE_PATH, instanceConfig.getInstanceId());
                logger.info("RegisterClient registerPath: {}", registerPath);
            } else {
                registerInstance();
            }
        } catch (Exception e) {
            logger.error("reRegisterInstance Exception : ", e);
        }
    }

    void registerInstance() {
        try {
            InstanceModel instanceModel = new InstanceModel();
            instanceModel.setInstanceId(instanceConfig.getInstanceId());
            instanceModel.setLoadCount(0l);
            String instanceModelStr = gson.toJson(instanceModel);
            registerPath = this.createEphemeralSequentialNode(Constants.INSTANCE_INSTANCE_PATH, instanceModelStr);
            logger.info("RegisterClient registerPath: {}", registerPath);
        } catch (Exception e) {
            logger.error("registerInstance Exception : ", e);
        }

    }

    void addInstanceListener() {
        PathChildrenCache cache = new PathChildrenCache(client, Constants.INSTANCE_PARENT_PATH, true);
        try {
            cache.start();
            cache.getListenable().addListener(new InstanceListener(this));
        } catch (Exception e) {
            logger.error("addInstanceListener Exception : ", e);
        }

    }

    public void updateInstanceInfo() {
        try {
            List<String> instances = this.getChildren(Constants.INSTANCE_PARENT_PATH);
            logger.info("instances :{}", instances);
            if (instanceModels.size() > 0) {
                instanceModels.clear();
            }
            for (String instance : instances) {

                String instancePath = Constants.INSTANCE_PARENT_PATH + "/" + instance;
                String instanceModelStr = this.getNodeData(instancePath);
                logger.info("instanceModelStr :{}", instanceModelStr);
                InstanceModel instanceModel = gson.fromJson(instanceModelStr, InstanceModel.class);
                instanceModels.add(instanceModel);
            }

        } catch (Exception e) {
            logger.error("updateInstanceInfo Exception : ", e);
        }

    }

    /**
     * 查询节点数据
     *
     * @param path
     * @return
     */
    public String getNodeData(String path) throws Exception {
        byte[] bytes = client.getData().forPath(path);
        return new String(bytes);
    }

    /**
     * 查询子节点
     *
     * @param path
     * @return
     */
    public List<String> getChildren(String path) throws Exception {
        List<String> workers = client.getChildren().forPath(path);
        return workers;

    }

    public Stat getNodeStat(String path) throws Exception {
        Stat stat = new Stat();
        client.getData()
                .storingStatIn(stat)
                .forPath(path);
        return stat;

    }

    public boolean isNodeExist(String path) throws Exception {
        return client.checkExists().forPath(path) != null;
    }

    public CuratorFramework getClient() {
        return client;
    }

    public void setClient(CuratorFramework client) {
        this.client = client;
    }


    public void sync() {
        client.sync();
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    public List<InstanceModel> getInstanceModels() {
        return instanceModels;
    }

    public void setInstanceModels(List<InstanceModel> instanceModels) {
        this.instanceModels = instanceModels;
    }

    /**
     * 注销
     */
    @PreDestroy
    private void cancel() {

        try {
            if (StringUtils.hasText(registerPath)) {
                this.deleteNode(registerPath);
            }
        } catch (Exception e) {
            logger.error("close Exception", e);
        }
    }
}
